package com.atguigu.day02

import com.atguigu.source.{SensorReading, SensorSource}
import org.apache.flink.streaming.api.scala._

object KeyedStreamExample {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    val stream: DataStream[SensorReading] = env.addSource(new SensorSource)


    //泛型变成了两个，第二个泛型为key的类型
    val keyed: KeyedStream[SensorReading, String] = stream.keyBy(_.id)

    //使用的三个字段来做滚动聚合，求每个传感器流上温度最小值
    //内部会保存一个最小值的状态变量，用来保存温度最小值
//    keyed.min(2).print()

    //reduce也会保存一个状态变量
    keyed.reduce((r1,r2)=>(
      SensorReading(r1.id,0,r1.temperature.min(r2.temperature))
    )).print()




    env.execute()

  }
}
